package com.bridgeintelligent.tag.customer.scheduled;


import cn.hutool.core.util.RandomUtil;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.bridgeintelligent.tag.constants.bulkload.MQConfig;
import com.bridgeintelligent.tag.constants.bulkload.StaticMqMsg;
import com.bridgeintelligent.tag.customer.mapper.CustomersMapper;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.wayneleo.quickstart.QuickException;
import lombok.AllArgsConstructor;
import lombok.SneakyThrows;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.stereotype.Component;

import java.util.Arrays;
import java.util.List;

import static com.bridgeintelligent.tag.constants.PublicConstant.THREE;
import static com.bridgeintelligent.tag.customer.constant.PublicConstant.ZERO;
import static com.bridgeintelligent.tag.customer.exception.CustomerErrors.ERROR_401006;

/**
 * 定时为静态客群向Hdfs写备份失败的操作二次执行
 */
@AllArgsConstructor
@Component
public class StaticCustomersScheduled {

    private CustomersMapper mapper;
    private ObjectMapper objectMapper;
    private RocketMQTemplate rocketMQTemplate;

    @SneakyThrows
    //@Scheduled(cron = "0 0/30 * * * ? ")//每30min执行一次
    public void doStaticHdfsWork() {
        //获取准备处理和失败的任务列表
        List<String> customerIds = mapper.findFailTask(Arrays.asList(ZERO,THREE));
        for (String customerId: customerIds) {
            //向MQ发送写HDFS的消息
            StaticMqMsg dynamicStaticMqMsg = new StaticMqMsg(customerId,mapper.findFlag(),customerId+"_"+ RandomUtil.randomString(6));
            try {
                String msg = objectMapper.writeValueAsString(dynamicStaticMqMsg);
                rocketMQTemplate.convertAndSend(MQConfig.STATIC_CUSTOMERS_HDFS_TOPIC, msg);
            } catch (JsonProcessingException e) {
                throw new RuntimeException(e);
            }catch(Exception e1){
                throw QuickException.normal(ERROR_401006.args(customerId));
            }
        }
    }

}
